package com.raymond.algorithm.statistical;

import org.apache.lucene.util.RamUsageEstimator;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.security.SecureRandom;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 统计每秒的计数
 *
 * @author :  raymond
 * @version :  V1.0
 * @date :  2022-08-23 10:43
 */

public class SpeedCounter {
    private final static Logger log = LoggerFactory.getLogger(SpeedCounter.class);
    /**
     * 多少毫秒一个小单位，默认1000毫秒，也就是1秒
     */
    private int scope;
    /**
     * 总共保留几个单位的量,默认三个单位
     */
    private int total;


    private WindowWrap[] slots;


    public SpeedCounter() {
        this(1000, 5);
    }

    public SpeedCounter(int scope, int total) {
        this.scope = scope;
        this.total = total;
        this.slots = new WindowWrap[this.total];
        for (int i = 0; i < this.slots.length; i++) {
            this.slots[i] = new WindowWrap();
        }
    }

    public int addCount() {
        return addCount(1);
    }

    public int addCount(int size) {
        long now = System.currentTimeMillis();
        int index = getIndex(now);
        addWaitingRequest(index, now);
        return slots[index].count.addAndGet(size);
    }

    /**
     * 获取小窗口的时间
     * @param timeMillis 时间戳
     * @return 小窗口的时间
     */
    private long calculateWindowStart(long timeMillis) {
        return timeMillis / scope;
    }

    private void addWaitingRequest(int index, long start) {
        start = calculateWindowStart(start);
        if (start > slots[index].time) {
            synchronized (this){
                if (start > slots[index].time) {
                    resetWindowTo(start);
                    slots[index].time = start;
                    slots[index].count.set(0);
                }
            }
        }
    }

    private void resetWindowTo(long start) {
        for (WindowWrap slot : slots) {
            if (start - slot.time >= total) {
                slot.count.set(0);
            }
        }
    }

    private int getIndex(long now) {
        return (int) ((now / scope) % total);
    }

    public int getTotalCount() {
        long now = System.currentTimeMillis();
        int index = getIndex(now);
        addWaitingRequest(index, now);
        int total = 0;
        for (WindowWrap slot : slots) {
            total += slot.count.get();
        }
        return total;
    }

    public int getAvg() {
        return getTotalCount() / this.slots.length;
    }

    public int getLastCount(int size) {
        if (size < 0 || size > this.slots.length) {
            throw new IllegalArgumentException("超过当前计数范围,当前计数范围:" + this.slots.length + ",size:" + size);
        }
        long now = System.currentTimeMillis();
        int index = getIndex(now);
        addWaitingRequest(index, now);
        if (index - size < 0) {
            index = this.slots.length + index - size;
        } else {
            index = index - size;
        }
        return this.slots[index].count.get();
    }

    public int getLastAvg(int size) {
        int total = getLastTotalCount(size);
        return total / size;
    }

    public int getLastTotalCount(int size) {
        if (size <= 0 || size > this.slots.length) {
            throw new IllegalArgumentException("超过当前计数范围,当前计数范围:" + this.slots.length + ",size:" + size);
        }
        long now = System.currentTimeMillis();
        int index = getIndex(now);
        addWaitingRequest(index, now);
        int total = 0;
        for (int i = 0; i < size; i++) {
            if (index == 0) {
                index = this.slots.length - 1;
            } else {
                index--;
            }
            total += this.slots[index].count.get();
        }
        return total;
    }

    private class WindowWrap {
        /**
         * 时间
         */
        private long time;
        /**
         * 请求次数
         */
        private AtomicInteger count = new AtomicInteger();
    }


    private static void test01() {
        SecureRandom secureRandom = new SecureRandom();
        SpeedCounter speedCounterCache = new SpeedCounter();
        ScheduledThreadPoolExecutor scheduledThreadPoolExecutor = new ScheduledThreadPoolExecutor(5);
        int threadNum = 16;
        long st = System.currentTimeMillis();
        long persistenceTime = 60 * 60 * 1000L;//持续60分钟
        AtomicInteger allCnt0 = new AtomicInteger();

        //创建多个线程模拟计数
        for (int i = 0; i < threadNum; i++) {
            Thread thread = new Thread(() -> {
                while (System.currentTimeMillis() - st < persistenceTime) {
                    try {
                        speedCounterCache.addCount();
                        allCnt0.incrementAndGet();
                        TimeUnit.MILLISECONDS.sleep(secureRandom.nextInt(5));
                    } catch (InterruptedException e) {
                        log.error("线程休眠被打断异常：", e);
                    }
                }
            });
            thread.setName("consumer_" + i);
            thread.start();
        }
        long total = 0;
        //使用一个定时任务不断输出
        scheduledThreadPoolExecutor.scheduleWithFixedDelay(new Runnable() {
            @Override
            public void run() {
                try {
                    int speedLast0 = speedCounterCache.getLastCount(0);
                    int speedLast1 = speedCounterCache.getLastCount(1);
                    int speedLast2 = speedCounterCache.getLastCount(2);
                    int speedLast3 = speedCounterCache.getLastCount(3);
                    int speedAvg3Seconds = speedCounterCache.getLastAvg(3);

                    log.warn("当前共发送了{}条，当前计数{}条,当前秒速为:{}，上一秒速为：{}，上两秒速为：{}，上三秒速为：{}，前三秒的平均速度为：{}",
                            0, allCnt0.get(), speedLast0, speedLast1, speedLast2, speedLast3, speedAvg3Seconds);
                    log.warn("当前java类的内存占用为：{}byte", RamUsageEstimator.sizeOf(speedCounterCache));
                } catch (Exception e) {
                    log.error("计算速度异常：", e);
                }
            }
        }, 0, 1, TimeUnit.SECONDS);
    }


    public static void main(String[] args) {
        test01();
    }

}
